/**
 * 
 */
package cc.rico.redis.utils.lock.impl;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import cc.rico.redis.utils.lock.RedisLockHandler;

/**
 * redis 并发锁
 * @author rico 2016年7月5日
 */
public class RedisTemplateLockHandlerImpl implements RedisLockHandler {
	private static final Logger logger = LoggerFactory.getLogger(RedisTemplateLockHandlerImpl.class);
	
	// 单个锁、批量锁获取超时时间(sec)
	private static final int DEFAULT_SINGLE_KEY_EXPIRED_TIME = 20;
	private static final int DEFAULT_BATCH_KEY_EXPIRED_TIME = 40;
	// 获取锁等待时间
	private static final int DEFAULT_LOCK_THREAD_SLEEP_TIME = 200;
	
	private RedisTemplate<String, Object> redisTemplate;

	/* 根据key获取锁
	 * @see cc.rico.redis.utils.lock.RedisLockHandler#tryLock(java.lang.String)
	 */
	@Override
	public boolean tryLock(String key) {
		return this.tryLock(key, 0, TimeUnit.SECONDS);
	}

	/* 根据key获取锁
	 * @see cc.rico.redis.utils.lock.RedisLockHandler#tryLock(java.lang.String, long, java.util.concurrent.TimeUnit)
	 */
	@Override
	public boolean tryLock(String key, long timeout, TimeUnit unit) {
		try {
			// 获取系统计时器的当前值，以毫微秒为单位。  
            long startNano = System.nanoTime(); 
            
			do {
				//long result = jedis.setnx(key, key);
				boolean result = redisTemplate.opsForValue().setIfAbsent(key, key);
				
				if(result) {
					//jedis.expire(key, DEFAULT_SINGLE_KEY_EXPIRED_TIME);
					redisTemplate.expire(key, DEFAULT_SINGLE_KEY_EXPIRED_TIME, TimeUnit.SECONDS);
					logger.info("RedisLockHandlerImpl.tryLock获取锁成功[key={}]", key);
					
					return true;
				} else {
					logger.info("RedisLockHandlerImpl.tryLock获取锁失败[key={}]", key);
				}
				
				// 未设置超时时间,直接返回获取锁失败
				if(timeout <= 0) {
					return false;
				}
				
				Thread.sleep(DEFAULT_LOCK_THREAD_SLEEP_TIME);
				
			} while(System.nanoTime()-startNano < unit.toNanos(timeout));
			
		} catch (InterruptedException e) {
			logger.error("根据key获取锁发生错误.", e);
		} finally {
			//jedis.close();
		}
		return false;
	}

	/* 根据keys timeout时间批量获取锁
	 * @see cc.rico.redis.utils.lock.RedisLockHandler#tryLock(java.util.List, long, java.util.concurrent.TimeUnit)
	 */
	@Override
	public boolean tryLock(List<String> keys, long timeout, TimeUnit unit) {
		
		try {
			// 需要获取的锁  
            List<String> requiredLockList = new CopyOnWriteArrayList<String>();  
            // 以获取的锁  
            List<String> lockedList = new CopyOnWriteArrayList<String>(); 
            
			// 获取系统计时器的当前值，以毫微秒为单位。  
            long startNano = System.nanoTime(); 
            
            do {
            	//Pipeline pipeline = jedis.pipelined();
            	/*for(String key : keys) {
            		requiredLockList.add(key);
            		pipeline.setnx(key, key);
            	}
            	List<Object> results = pipeline.syncAndReturnAll();
            	*/
            	
            	RedisCallback<List<Object>> pipelineCallback = new RedisCallback<List<Object>>(){  
					@Override
					public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
						connection.openPipeline();
						
						for(String key : keys) {
							requiredLockList.add(key);
							
							byte[] rawKey = redisTemplate.getStringSerializer().serialize(key);
							connection.setNX(rawKey, rawKey);
						}
						return connection.closePipeline();
					}  
                };  
                
                List<Object> results = redisTemplate.executePipelined(pipelineCallback);
            	for(int i=0; i<results.size(); i++) {
            		boolean ret = ((Boolean) results.get(i)).booleanValue();
            		String key = requiredLockList.get(i);
            		
            		if(ret) {
            			//jedis.expire(key, DEFAULT_BATCH_KEY_EXPIRED_TIME);
            			redisTemplate.expire(key, DEFAULT_BATCH_KEY_EXPIRED_TIME, TimeUnit.SECONDS);
            			lockedList.add(key);
            		}
            	}
            	
            	requiredLockList.removeAll(lockedList);
            	
            	if(requiredLockList.size() == 0) {
            		logger.info("已成功获取全部锁[lockedList={}]", lockedList);
            		return true;
            	} else {
            		logger.info("未获取全部锁[requiredLockList={}]", requiredLockList);
            	}
            	
            	// 未设置获取锁超时时间，直接返回获取锁失败
            	if(timeout <= 0) {
            		break;
            	}
            	
            	Thread.sleep(DEFAULT_LOCK_THREAD_SLEEP_TIME);
            	
            } while(System.nanoTime()-startNano < unit.toNanos(timeout));
            
            // 未获取全部锁,释放以获取的锁
            if(lockedList.size() > 0) {
            	//jedis.del(lockedList.toArray(new String[lockedList.size()]));
            	redisTemplate.delete(lockedList);
            }
            
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			//jedis.close();
		}
		
		return false;
	}

	/* 根据key获取锁
	 * @see cc.rico.redis.utils.lock.RedisLockHandler#lock(java.lang.String)
	 */
	@Override
	public void lock(String key) {
		try {
            do {
            	//long result = jedis.setnx(key, key);
            	boolean result = redisTemplate.opsForValue().setIfAbsent(key, key);
            	
            	if(result) {
            		//jedis.expire(key, DEFAULT_SINGLE_KEY_EXPIRED_TIME);
            		redisTemplate.expire(key, DEFAULT_SINGLE_KEY_EXPIRED_TIME, TimeUnit.SECONDS);
            		logger.info("获取锁成功[key={}]", key);
            		
            		return;
            	} else {
            		logger.info("获取锁失败[key={}]", key);
            	}
            	
            	Thread.sleep(DEFAULT_LOCK_THREAD_SLEEP_TIME);
            	
            } while(true);
            
		} catch (InterruptedException e) {
			logger.error("根据key获取锁发生错误.", e);
		} finally {
			//jedis.close();
		}
	}

	/* 释放锁
	 * @see cc.rico.redis.utils.lock.RedisLockHandler#unlock(java.lang.String)
	 */
	@Override
	public void unlock(String key) {
		List<String> keys = Arrays.asList(key);
		this.unlock(keys);
	}

	/* 批量释放锁
	 * @see cc.rico.redis.utils.lock.RedisLockHandler#unlock(java.util.List)
	 */
	@Override
	public void unlock(List<String> keys) {
		try {
			//jedis.del(keys.toArray(new String[keys.size()]));
			redisTemplate.delete(keys);
			
		} finally {
			//jedis.close();
		}
		
	}
	
	public RedisTemplate<String, Object> getRedisTemplate() {
		return redisTemplate;
	}

	public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
		this.redisTemplate = redisTemplate;
	}
}